// Copyright 2019 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: listener code decoupling
// feature: dserver support type discovery
// feature: develop dispatcher
// feature: Split history as an independent common module
// feature: support config log level by cmd tool
// feature: add statictic lookup
// ------------------------------------------------------------------

#include <edds/dds/builtin/lookup/LookupManager.hpp>
#include <edds/dds/builtin/lookup/LookupRequestListener.hpp>
#include <history/WriterHistory.h>
#include <edds/rtps/participant/RTPSParticipant.h>
#include <ertps/types/TypeObjectFactory.h>
#include <ertps/rtps/reader/StatefulReader.h>
#include <ertps/rtps/history/ReaderHistory.h>
#include <rtps/participant/RTPSParticipantImpl.h>
#include <utility>
#include <elog/Log.hpp>
#include <deps/common/SystemInfo.hpp>

using evbs::ertps::rtps::c_EntityId_Lookup_request_writer;
using evbs::ertps::rtps::RTPSReader;
using evbs::ertps::rtps::RTPSWriter;
using evbs::ertps::types::TypeIdentifier;
using evbs::ertps::types::TypeIdentifierPair;
using evbs::ertps::types::TypeIdentifierTypeObjectPair;
using evbs::ertps::types::TypeIdentifierWithSize;
using evbs::ertps::types::TypeIdentifierWithSizeSeq;
using evbs::ertps::types::TypeObject;
using evbs::ertps::types::TypeObjectFactory;
using vbs::common::CacheChange_t;
using vbsutil::elog::Log;

namespace evbs {
namespace edds {
namespace dds {
namespace builtin {

LookupRequestListener::LookupRequestListener(LookupManager* manager)
    : tlm_(manager),
      factory_(TypeObjectFactory::get_instance()),
      get_writer_reader_infos_funcptr(nullptr),
      get_remote_guids_funcptr(nullptr),
      get_statistic_send_info_funcptr(nullptr),
      get_statistic_recv_info_funcptr(nullptr),
      get_statistic_writer_reader_qos_funcptr(nullptr),
      get_statistic_writer_reader_proxy_infos_funcptr(nullptr) {}

LookupRequestListener::~LookupRequestListener() {
    get_writer_reader_infos_funcptr = nullptr;
    get_remote_guids_funcptr = nullptr;
    get_statistic_send_info_funcptr = nullptr;
    get_statistic_recv_info_funcptr = nullptr;
    get_statistic_writer_reader_qos_funcptr = nullptr;
    get_statistic_writer_reader_proxy_infos_funcptr = nullptr;
}

void LookupRequestListener::getWriterReaderInfo(Lookup_Request& request) {
    Lookup_getWriterReaderInfo_Result out {};
    if (request.data.getGuidInfo().guid_ != tlm_->participant_->getGuid()) {
        if (tlm_->get_discovery_protocol() != ertps::rtps::DiscoveryProtocol::SERVER) {
            return;
        }
        // DServer转发功能, timeout需要比DClient小
        // 如果转发给Dclinet，dest_guid直接填请求的即可
        out.infos_ = tlm_->lookup_writer_reader_infos(request.data.getGuidInfo().guid_, (VBS_TOOL_WAIT_TIMEOUT / 2));
    } else {
        if (get_writer_reader_infos_funcptr != nullptr) {
            out.infos_ = get_writer_reader_infos_funcptr();
        }
    }
    Lookup_Reply* reply = static_cast<Lookup_Reply*>(tlm_->reply_type_.create_data());
    reply->data.getWriterReaderInfo(out);
    reply->header.requestId = request.header.requestId;

    (void)tlm_->send_reply(*reply);
    tlm_->reply_type_.delete_data(reply);
    logDebug(LOOKUP_SERVICE, request.data.getGuidInfo().guid_ << " Received writer reader info request.");
}

void LookupRequestListener::getMatchGuids(Lookup_Request& request) {
    Lookup_getMatchGuids_Result out {};
    if (request.data.getGuidInfo().guid_.guidPrefix != tlm_->participant_->getGuid().guidPrefix) {
        if (tlm_->get_discovery_protocol() != ertps::rtps::DiscoveryProtocol::SERVER) {
            return;
        }
        out.guids_ = tlm_->lookup_remote_guids(request.data.getGuidInfo().guid_, (VBS_TOOL_WAIT_TIMEOUT / 2));
    } else {
        if (get_remote_guids_funcptr != nullptr) {
            out.guids_ = get_remote_guids_funcptr(request.data.getGuidInfo().guid_);
        }
    }
    Lookup_Reply* reply = static_cast<Lookup_Reply*>(tlm_->reply_type_.create_data());
    reply->data.getMatchGuids(out);
    reply->header.requestId = request.header.requestId;

    (void)tlm_->send_reply(*reply);
    tlm_->reply_type_.delete_data(reply);
    logDebug(LOOKUP_SERVICE, request.data.getGuidInfo().guid_ << " Received match guids request.");
}

void LookupRequestListener::getProxyInfos(Lookup_Request& request) {
    Lookup_getProxyInfos_Result out;
    if (request.data.getGuidInfo().guid_.guidPrefix != tlm_->participant_->getGuid().guidPrefix) {
        if (tlm_->get_discovery_protocol() != ertps::rtps::DiscoveryProtocol::SERVER) {
            return;
        }
        out.proxy_infos_ =
            tlm_->lookup_writer_reader_proxy_infos(request.data.getGuidInfo().guid_, (VBS_TOOL_WAIT_TIMEOUT / 2));
    } else {
        if (get_statistic_writer_reader_proxy_infos_funcptr != nullptr) {
            out.proxy_infos_ = get_statistic_writer_reader_proxy_infos_funcptr(request.data.getGuidInfo().guid_);
        }
    }

    Lookup_Reply* reply = static_cast<Lookup_Reply*>(tlm_->reply_type_.create_data());
    reply->data.getProxyInfos(out);
    reply->header.requestId = request.header.requestId;

    (void)tlm_->send_reply(*reply);
    tlm_->reply_type_.delete_data(reply);
    logDebug(LOOKUP_SERVICE, request.data.getGuidInfo().guid_ << " Received match guids request.");
}

void LookupRequestListener::getSendInfo(Lookup_Request& request) {
    Lookup_getSendInfo_Result out {};
    if (request.data.getGuidInfo().guid_.guidPrefix != tlm_->participant_->getGuid().guidPrefix) {
        if (tlm_->get_discovery_protocol() != ertps::rtps::DiscoveryProtocol::SERVER) {
            return;
        }
        (void)tlm_->lookup_statistic_send_info(request.data.getGuidInfo().guid_, &out.info_,
                                               (VBS_TOOL_WAIT_TIMEOUT / 2));
    } else {
        if (get_statistic_send_info_funcptr != nullptr) {
            if (get_statistic_send_info_funcptr(request.data.getGuidInfo().guid_, &out.info_) !=
                ReturnCode_t::RETCODE_OK) {
                return;
            }
        }
    }

    Lookup_Reply* reply = static_cast<Lookup_Reply*>(tlm_->reply_type_.create_data());
    reply->data.getSendInfo(out);
    reply->header.requestId = request.header.requestId;

    (void)tlm_->send_reply(*reply);
    tlm_->reply_type_.delete_data(reply);
    logDebug(LOOKUP_SERVICE, request.data.getGuidInfo().guid_ << " Received send info request.");
}

void LookupRequestListener::getRecvInfo(Lookup_Request& request) {
    Lookup_getRecvInfo_Result out {};
    if (request.data.getGuidInfo().guid_.guidPrefix != tlm_->participant_->getGuid().guidPrefix) {
        if (tlm_->get_discovery_protocol() != ertps::rtps::DiscoveryProtocol::SERVER) {
            return;
        }
        (void)tlm_->lookup_statistic_recv_info(request.data.getGuidInfo().guid_, &out.info_,
                                               (VBS_TOOL_WAIT_TIMEOUT / 2));
    } else {
        if (get_statistic_recv_info_funcptr != nullptr) {
            if (get_statistic_recv_info_funcptr(request.data.getGuidInfo().guid_, &out.info_) !=
                ReturnCode_t::RETCODE_OK) {
                return;
            }
        }
    }

    Lookup_Reply* reply = static_cast<Lookup_Reply*>(tlm_->reply_type_.create_data());
    reply->data.getRecvInfo(out);
    reply->header.requestId = request.header.requestId;

    (void)tlm_->send_reply(*reply);
    tlm_->reply_type_.delete_data(reply);
    logDebug(LOOKUP_SERVICE, request.data.getGuidInfo().guid_ << " Received recv info request.");
}

void LookupRequestListener::getWriterQos(Lookup_Request& request) {
    Lookup_getWriterQos_Result out {};
    if (request.data.getGuidInfo().guid_.guidPrefix != tlm_->participant_->getGuid().guidPrefix) {
        if (tlm_->get_discovery_protocol() != ertps::rtps::DiscoveryProtocol::SERVER) {
            return;
        }
        (void)tlm_->lookup_statistic_writer_reader_qos(request.data.getGuidInfo().guid_, &out.qos_,
                                                       (VBS_TOOL_WAIT_TIMEOUT / 2));
    } else {
        if (get_statistic_writer_reader_qos_funcptr != nullptr) {
            if (get_statistic_writer_reader_qos_funcptr(request.data.getGuidInfo().guid_, &out.qos_) !=
                ReturnCode_t::RETCODE_OK) {
                return;
            }
        }
    }

    Lookup_Reply* reply = static_cast<Lookup_Reply*>(tlm_->reply_type_.create_data());
    reply->data.getWriterQos(out);
    reply->header.requestId = request.header.requestId;

    (void)tlm_->send_reply(*reply);
    tlm_->reply_type_.delete_data(reply);
    logDebug(LOOKUP_SERVICE, request.data.getGuidInfo().guid_ << " Received writer qos request.");
}

void LookupRequestListener::getMessageBrief(Lookup_Request& request) {
    Lookup_getMsgBrief_Result out {};
    std::string host_name = vbs::SystemInfo::get_hostname();
    const std::string& topic_name = request.data.remoteConfig().param_name;
    if (static_cast<pid_t>(request.data.remoteConfig().pid) != getpid()) {
        if ((tlm_->get_discovery_protocol() == ertps::rtps::DiscoveryProtocol::SERVER) &&
            (request.data.remoteConfig().host_name == host_name)) {
            out.guid_msg_briefs_ = tlm_->lookup_statistic_message_brief(request.data.remoteConfig().host_name,
                                                                        request.data.remoteConfig().pid, topic_name,
                                                                        (VBS_TOOL_WAIT_TIMEOUT / 2));
        } else {
            return;
        }
    } else {
        out.guid_msg_briefs_ = Log::GetMessageBrief(topic_name);
    }

    Lookup_Reply* reply = static_cast<Lookup_Reply*>(tlm_->reply_type_.create_data());
    reply->data.getMsgBreifInfo(out);
    reply->header.requestId = request.header.requestId;

    (void)tlm_->send_reply(*reply);
    tlm_->reply_type_.delete_data(reply);
    logDebug(LOOKUP_SERVICE, " Received message brief request. topic " << topic_name << " result count "
                                                                       << out.guid_msg_briefs_.size());
}

void LookupRequestListener::getTypes(Lookup_Request& request) {
    const Lookup_getTypes_Req in = request.data.getTypes();
    Lookup_getTypes_Result out {};

    for (const TypeIdentifier& type_id : in.type_ids) {
        TypeObject obj;
        const TypeIdentifier* obj_ident = factory_->typelookup_get_type(type_id, obj);
        if ((obj_ident == nullptr) || (obj._d() == 0U)) {
            return;
        }

        TypeIdentifierTypeObjectPair pair;
        pair.type_identifier(type_id);
        pair.type_object(obj);
        out.types.push_back(std::move(pair));

        if (!(type_id == *obj_ident)) {
            TypeIdentifierPair pair;
            pair.type_identifier1(*obj_ident);
            pair.type_identifier2(type_id);
            out.complete_to_minimal.push_back(std::move(pair));
        }
    }

    Lookup_Reply* reply = static_cast<Lookup_Reply*>(tlm_->reply_type_.create_data());
    reply->data.getType(out);
    reply->header.requestId = request.header.requestId;

    (void)tlm_->send_reply(*reply);
    tlm_->reply_type_.delete_data(reply);
    logDebug(LOOKUP_SERVICE, tlm_->get_builtin_request_writer_guid() << " Received types request.");
}

void LookupRequestListener::getDependencies(Lookup_Request& request) {
    const Lookup_getTypeDependencies_Req in = request.data.getTypeDependencies();
    Lookup_getTypeDependencies_Result out {};

    for (const TypeIdentifier& identifier : in.type_ids) {
        const TypeIdentifier* local_id = factory_->get_stored_type_identifier(&identifier);
        if (local_id == nullptr) {
            return;
        }
    }

    out.dependent_typeids = factory_->typelookup_get_type_dependencies(
        in.type_ids, in.continuation_point, out.continuation_point, 255U);  // TODO: Make configurable?

    Lookup_Reply* reply = static_cast<Lookup_Reply*>(tlm_->reply_type_.create_data());
    reply->data.getTypeDependencies(out);
    reply->header.requestId = request.header.requestId;

    (void)tlm_->send_reply(*reply);
    tlm_->reply_type_.delete_data(reply);
    logDebug(LOOKUP_SERVICE, tlm_->get_builtin_request_writer_guid() << " Received type dependencies request.");
}

void LookupRequestListener::remoteConfig(Lookup_Request& request) {
    Lookup_remoteConfig_Result out {};
    std::string host_name = vbs::SystemInfo::get_hostname();
    if (request.data.remoteConfig().host_name != host_name ||
        static_cast<pid_t>(request.data.remoteConfig().pid) != getpid()) {
        if ((tlm_->get_discovery_protocol() == ertps::rtps::DiscoveryProtocol::SERVER) &&
            (request.data.remoteConfig().host_name == host_name)) {
            out.result_ = tlm_->remoteConfig(host_name, request.data.remoteConfig().pid,
                                             request.data.remoteConfig().param_name, (VBS_TOOL_WAIT_TIMEOUT / 2))();
        } else {
            return;
        }
    } else {
        const std::string& params = request.data.remoteConfig().param_name;
        out.result_ = LookupManager::localConfig(params)();
    }

    Lookup_Reply* reply = static_cast<Lookup_Reply*>(tlm_->reply_type_.create_data());
    reply->data.remoteConfig(out);
    reply->header.requestId = request.header.requestId;

    (void)tlm_->send_reply(*reply);
    tlm_->reply_type_.delete_data(reply);
    logDebug(LOOKUP_SERVICE,
             " Received host process cmd request. host name " << host_name << " result " << out.result_);
}

void LookupRequestListener::getDServerInfo(Lookup_Request& request) {
    Lookup_getDServerInfo_Result out {};
    std::string host_name = vbs::SystemInfo::get_hostname();
    if (request.data.remoteConfig().host_name != host_name ||
        (tlm_->get_discovery_protocol() != ertps::rtps::DiscoveryProtocol::SERVER)) {
        return;
    }

    out.infos_ = tlm_->participant_->get_db_info(false);

    Lookup_Reply* reply = static_cast<Lookup_Reply*>(tlm_->reply_type_.create_data());
    reply->data.dserverInfo(out);
    reply->header.requestId = request.header.requestId;

    (void)tlm_->send_reply(*reply);
    tlm_->reply_type_.delete_data(reply);
    logDebug(LOOKUP_SERVICE, " Received host cmd request for dserver info. host name " << host_name);
}

void LookupRequestListener::onNewCacheChangeAdded(vbs::dispatcher::SubstrateReader* sub_reader,
                                                  const CacheChange_t* const changeIN) {
    logDebug(LOOKUP_SERVICE, "Received LookupRequestListener.");
    CacheChange_t* change = const_cast<CacheChange_t*>(changeIN);
    RTPSReader* reader = static_cast<RTPSReader*>(sub_reader);
    if (change->writerGUID.entityId != c_EntityId_Lookup_request_writer) {
        elogWarning(LOOKUP_SERVICE, "Received data from a bad endpoint.");
        (void)reader->getHistory()->remove_change(change);
        return;
    }

    // 1.4.0版本及后续版本和1.3.0版本的类型发现不再兼容
    const auto& version = tlm_->participant_->get_version_by_guid(change->writerGUID.guidPrefix);
    if (version[0] == 1 && version[1] == 3 && version[2] == 0) {
        elogWarning(LOOKUP_SERVICE,
                    "It is no longer compatible with version " << version[0] << "." << version[1] << "." << version[2]);
        (void)reader->getHistory()->remove_change(change);
        return;
    }

    Lookup_Request request;
    if (tlm_->recv_request(*change, request)) {
        if (request.header.requestId.writer_guid() == tlm_->get_builtin_request_writer_guid()) {
            // Message from our selves.
            (void)reader->getHistory()->remove_change(change);
            return;
        }

        switch (request.data._d()) {
            case Statistic_getWRInfo_Hash:
                getWriterReaderInfo(request);
                break;
            case Statistic_getMatchGuids_Hash:
                getMatchGuids(request);
                break;
            case Statistic_getSendInfo_Hash:
                getSendInfo(request);
                break;
            case Statistic_getRecvInfo_Hash:
                getRecvInfo(request);
                break;
            case Statistic_getWriterReaderQos_Hash:
                getWriterQos(request);
                break;
            case Statistic_getMsgBrief_Hash:
                getMessageBrief(request);
                break;
            case Statistic_getProxyInfo_Hash:
                getProxyInfos(request);
                break;
            case Statistic_remoteConfig_Hash:
                remoteConfig(request);
                break;
            case Statistic_getDServerInfo_Hash:
                getDServerInfo(request);
                break;
            case Lookup_getTypeInfo_Hash: {
                tlm_->type_info_reply(request.data.getTypeInfo().type_name, request.header.requestId);
                break;
            }
            case Lookup_getTypes_Hash: {
                if (request.header.dest_guid == tlm_->get_builtin_reply_writer_guid()) {
                    getTypes(request);
                }
                break;
            }
            case Lookup_getDependencies_Hash: {
                if (request.header.dest_guid == tlm_->get_builtin_reply_writer_guid()) {
                    getDependencies(request);
                }
                break;
            }
            default:
                break;
        }
    }
    (void)reader->getHistory()->remove_change(change);
}

void LookupRequestListener::onWriterChangeReceivedByAll(vbs::dispatcher::SubstrateWriter* const /*writer*/,
                                                        CacheChange_t* change) {
    tlm_->builtin_reply_writer_history_->remove_change(change);
}

}  // namespace builtin
}  // namespace dds
}  // namespace edds
}  // namespace evbs
